package com.superbytecode.cloud.rocketmq.consumer.provider;

import com.superbytecode.cloud.common.model.mq.MQTopic;
import com.superbytecode.cloud.common.utils.VerifyUtil;
import com.superbytecode.cloud.rocketmq.consumer.Consumer;
import com.superbytecode.cloud.rocketmq.configure.MQProviderEnum;
import com.superbytecode.cloud.rocketmq.consumer.MQConsumeProvider;
import com.superbytecode.cloud.rocketmq.exception.MQBizException;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.AccessChannel;
import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.autoconfigure.RocketMQProperties;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQReplyListener;
import org.apache.rocketmq.spring.support.DefaultRocketMQListenerContainer;
import org.apache.rocketmq.spring.support.RocketMQMessageConverter;
import org.springframework.beans.factory.support.BeanDefinitionValidationException;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.support.GenericApplicationContext;
import org.springframework.core.ResolvableType;
import org.springframework.core.env.StandardEnvironment;
import sun.reflect.annotation.AnnotationParser;
import sun.reflect.annotation.AnnotationType;

import java.lang.reflect.Field;
import java.lang.reflect.Type;
import java.util.Collections;
import java.util.Map;
import java.util.Objects;

/**
 * All rights Reserved, Designed By www.super-master.com
 *
 * @ProjectName: beacon-cloud
 * @Package: com.superbytecode.cloud.rocketmq.consumer
 * @ClassName: OpenMQConsumeContainer
 * @Description: [开源版MQ消费提供服务]
 * <p> </p>
 * @Author: [Lance Ting]
 * @Date: 2024/11/1 17:53
 * @Version: V1.0
 * @Copyright: 2024 www.super-master.com Inc. All rights reserved.
 * TODO: 注意,本文件Lance Ting所作,如果转载或使用请标明具体出处!
 **/
@Slf4j
@AllArgsConstructor
public class OpenMQConsumeProvider implements MQConsumeProvider {
    private final ConfigurableApplicationContext context;
    private final StandardEnvironment environment;
    private final RocketMQProperties properties;
    private final RocketMQMessageConverter converter;

    /**
     * 注册容器
     *
     * @param targetClass      目标类
     * @param listenerBeanName bean名称
     * @param topicBean        消费者的bean
     * @param topic            topic名字
     * @param group            消费组id
     * @param consumer         消费者配置
     */
    @Override
    public void register(Class<?> targetClass, String listenerBeanName, Object topicBean, String topic, String group, Consumer consumer) {
        Class<DefaultRocketMQListenerContainer> listenerContainerClass = DefaultRocketMQListenerContainer.class;
        String tag = consumer.tag();
        ConsumeMode consumeMode = consumer.consumeMode();
        MessageModel messageModel = consumer.messageModel();
        // 解析RocketMQMessageListener注解
        RocketMQMessageListener listenerAnnotation = annotationForMessageListener(topic, group, tag, consumeMode, messageModel);
        boolean listenerEnabled = properties.getConsumer().getListeners() //
                .getOrDefault(group, Collections.emptyMap()) //
                .getOrDefault(topic, true);
        if (!listenerEnabled) {
            log.debug("Consumer Listener (group:{},topic:{}) is not enabled by configuration, will ignore initialization.", group, topic);
            return;
        }
        // 验证
        validate(listenerAnnotation);
        String containerBeanName = listenerContainerClass.getName();
        try {
            GenericApplicationContext applicationContext = (GenericApplicationContext) context;
            @SuppressWarnings("unchecked")//
            MQTopic<Object> messageContext = (MQTopic<Object>) topicBean;
            RocketMQListener<Object> rocketListener = messageContext::consume;

            // 注册容器并获取容器实例
            applicationContext.registerBean(containerBeanName, listenerContainerClass,//
                    () -> this.createMessageListenerContainer(containerBeanName, rocketListener, listenerAnnotation)//
            );
            DefaultRocketMQListenerContainer container = applicationContext.getBean(listenerBeanName, listenerContainerClass);
            if (container.isRunning()) {
                return;
            }
            // 运行容器
            ResolvableType[] resolvableInterfaces = ResolvableType.forClass(targetClass).getInterfaces();
            ResolvableType resolvableGenericType = resolvableInterfaces[0].getInterfaces()[0].getGeneric(0);
            Type resolvableType = resolvableGenericType.getType();
            // 
            Field messageType = container.getClass().getDeclaredField("messageType");
            messageType.setAccessible(true);
            messageType.set(container, resolvableType);
        } catch (NoSuchFieldException | IllegalAccessException e) {
            throw new MQBizException(e);
        } finally {
            log.info("Register the listener to container, listenerBeanName:{}, containerBeanName:{}", listenerBeanName, containerBeanName);
        }
    }

    /**
     * QM服务提供者
     *
     * @return 阿里MQ|开源MQ
     */
    @Override
    public MQProviderEnum provider() {
        return MQProviderEnum.OPEN;
    }

    /**
     * 创建DefaultRocketMQListenerContainer对象
     *
     * @param containerBeanName container名称
     * @param container         container对象
     * @param annotation        RocketMQMessageListener注解
     * @return org.apache.rocketmq.spring.support.DefaultRocketMQListenerContainer 实例
     */
    private DefaultRocketMQListenerContainer createMessageListenerContainer(String containerBeanName, Object container, RocketMQMessageListener annotation) {
        DefaultRocketMQListenerContainer listenerContainer = new DefaultRocketMQListenerContainer();
        listenerContainer.setRocketMQMessageListener(annotation);
        //
        String nameServer = this.resolveEnvironmentValue(annotation.nameServer(), properties.getNameServer());
        listenerContainer.setNameServer(nameServer);
        String accessChannel = this.resolveEnvironmentValue(annotation.accessChannel(), "");
        if (VerifyUtil.isNotEmpty(accessChannel)) {
            listenerContainer.setAccessChannel(AccessChannel.valueOf(accessChannel));
        }
        String topic = this.resolveEnvironmentValue(annotation.topic(), "");
        listenerContainer.setTopic(topic);
        String tags = this.resolveEnvironmentValue(annotation.selectorExpression(), "");
        if (!StringUtils.isEmpty(tags)) {
            listenerContainer.setSelectorExpression(tags);
        }
        String consumerGroup = this.resolveEnvironmentValue(annotation.consumerGroup(), "");
        listenerContainer.setConsumerGroup(consumerGroup);
        // 老版本没有这个字段，暂时先注释掉
        if (RocketMQListener.class.isAssignableFrom(container.getClass())) {
            listenerContainer.setRocketMQListener((RocketMQListener<?>) container);
        } else if (RocketMQReplyListener.class.isAssignableFrom(container.getClass())) {
            listenerContainer.setRocketMQReplyListener((RocketMQReplyListener<?, ?>) container);
        }
        listenerContainer.setMessageConverter(converter.getMessageConverter());
        listenerContainer.setName(containerBeanName);

        return listenerContainer;
    }

    private String resolveEnvironmentValue(String value, String defaultValue) {
        String resolveValue = environment.resolvePlaceholders(value);
        if (StringUtils.isEmpty(defaultValue)) {
            return resolveValue;
        }
        return VerifyUtil.isNull(resolveValue) ? defaultValue : resolveValue;
    }

    /**
     * 映射MQ监听器对象
     *
     * @param topic         topic名字
     * @param consumerGroup 消费组
     * @param tag           消息标签
     * @param consumeMode   消费模式：顺序/并发
     * @param messageModel  消息模式: 集群/广播
     * @return 监听器
     */
    private RocketMQMessageListener annotationForMessageListener(String topic, String consumerGroup, String tag,//
                                                                 ConsumeMode consumeMode, MessageModel messageModel) {
        // 获取注解默认属性
        AnnotationType instance = AnnotationType.getInstance(RocketMQMessageListener.class);
        Map<String, Object> stringObjectMap = instance.memberDefaults();
        // 完善注解必填属性
        stringObjectMap.put("topic", topic);
        stringObjectMap.put("consumerGroup", consumerGroup);
        stringObjectMap.put("selectorExpression", tag);
        stringObjectMap.put("consumeMode", consumeMode);
        stringObjectMap.put("messageModel", messageModel);
        // 反射生成实例
        return (RocketMQMessageListener) AnnotationParser.annotationForMap(RocketMQMessageListener.class, stringObjectMap);
    }

    private void validate(RocketMQMessageListener annotation) {
        ConsumeMode consumeMode = annotation.consumeMode();
        MessageModel messageModel = annotation.messageModel();
        if (Objects.equals(consumeMode, ConsumeMode.ORDERLY) && Objects.equals(messageModel, MessageModel.BROADCASTING)) {
            throw new BeanDefinitionValidationException(//
                    "Bad annotation definition in @RocketMQMessageListener, messageModel BROADCASTING does not support ORDERLY message!");
        }
    }
}